﻿using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using GeneralDFS.BusinessLogical.IO;
using GeneralDFS.BusinessLogical.Network;
using GeneralDFS.Common;
using GeneralDFS.DataEntity.Enum;
using GeneralDFS.DataEntity.IO;
using GeneralDFS.BusinessLogical.Protocol;

namespace GeneralDFS.BusinessLogical.Nodes
{
    public  partial class DataNode
    {
        private TcpListener _tcpListener;
        private readonly BackgroundWorker _backgroundTcpListener=new BackgroundWorker();

        public void InitListen()
        {
            LogUtil.LogInfo("----------Starting init DataNodeTcpListener------------");
            _tcpListener = new TcpListener(IPAddress.Parse(_dataNodeConfig.Ip), _dataNodeConfig.TcpPort);
            LogUtil.LogInfo(String.Format("Ip:{0} TcpPort:{1}", _dataNodeConfig.Ip, _dataNodeConfig.TcpPort));
            _backgroundTcpListener.DoWork += _backgroundTcpListener_DoWork;
            _backgroundTcpListener.RunWorkerAsync();
            LogUtil.LogInfo("----------init DataNodeTcpListener Success------------");
        }

        private void _backgroundTcpListener_DoWork(object sender, DoWorkEventArgs e)
        {
            _tcpListener.Start(1000);
            var worker = (BackgroundWorker) sender;
            var accpetDone = new ManualResetEvent(false);

            while (!worker.CancellationPending)
            {
                accpetDone.Reset();
                var acceptState = new CustomTcpState() { TcpListener = _tcpListener, AllDone = accpetDone };
                try
                {
                    _tcpListener.BeginAcceptTcpClient(AcceptTcpClientCallback, acceptState);
                    acceptState.AllDone.WaitOne();
                }
                catch (Exception ex)
                {
                    LogUtil.LogError(ex);
                    acceptState.AllDone.Reset();
                }
            }
        }

        private void AcceptTcpClientCallback(IAsyncResult result)
        {
            var state = result.AsyncState as CustomTcpState;
            Debug.Assert(state != null);
            state.TcpClient = state.TcpListener.EndAcceptTcpClient(result);
            state.NetworkStream = state.TcpClient.GetStream();
            state.AllDone.Set();
            var binaryReader=new BinaryReader(state.NetworkStream,Encoding.UTF8);
            DateTime beginReadOp = DateTime.Now;
            var op = ProcessOp.ReadOp(binaryReader);
            Console.WriteLine("ReadOpTime {0}",(DateTime.Now-beginReadOp).TotalSeconds);
            byte[] data;
            string filePath;
            Guid guid;
            DateTime begin;
            DateTime end;
            FileBlockMetaData fileBlockMetaData;
            switch (op)
            {
                case Op.ReadBlock:
                    DataTranslater.ReadHead(new BinaryReader(state.NetworkStream), out guid);
                    ReadBlockMetaData(out fileBlockMetaData, guid);
                    filePath = FSVolume.GetInstance().GetFilePath(guid);
                    begin = DateTime.Now;
                    DataTranslater.WriteBlock(fileBlockMetaData, new BinaryWriter(state.NetworkStream), System.IO.File.OpenRead(filePath), Config.DetailConfig.BuffSize);
                    end = DateTime.Now;
                    Speed.GetInstance().UpdateSpeed(end - begin, fileBlockMetaData.Length);
                    state.NetworkStream.Dispose();
                    break;
                case Op.WriteBlock:
                    var tmpFile = Path.Combine(_dataNodeConfig.DataNodeDirectory, DFSUtils.GetRandomString(10));
                    var stream = System.IO.File.Create(tmpFile);
                    begin = DateTime.Now;
                    DataTranslater.ReadBlock(out fileBlockMetaData,binaryReader,stream,Config.DetailConfig.BuffSize);
                    end = DateTime.Now;
                    Speed.GetInstance().UpdateSpeed(end - begin, fileBlockMetaData.Length);
                    //fileBlockMetaData.Hash = DataTranslater.WriteHash(state.NetworkStream, stream);
                    if(stream.CanRead)
                        stream.Dispose();
                    state.NetworkStream.Dispose();
                    ProcessReadBlock(fileBlockMetaData, tmpFile);
                    break;
                    
                case Op.ReadSmallBlock:
                    DataTranslater.ReadHead(new BinaryReader(state.NetworkStream), out guid);
                    begin = DateTime.Now;
                    data = FileData.GetInstance().GetSmallFileData(guid);
                    DataTranslater.ClientRequestSmallBlock(new BinaryWriter(state.NetworkStream),data);
                    end = DateTime.Now;
                    //Speed.GetInstance().UpdateSpeed(end-begin,data.Length);
                    state.NetworkStream.Dispose();
                    break;

                case Op.WriteSmallBlock:
                    begin = DateTime.Now;
                    data = DataTranslater.ClientSendSmallBlock(out fileBlockMetaData,binaryReader);
                    end = DateTime.Now;
                    Speed.GetInstance().UpdateSpeed(end - begin, fileBlockMetaData.Length);
                    state.NetworkStream.Dispose();
                    IO.FileData.GetInstance().AddFileToFileData(data, fileBlockMetaData);
                    break;
            }
        }

        private void ProcessReadBlock(FileBlockMetaData fileBlockMetaData, String tmpFile)
        {
            //var blockFilePath = Path.Combine(_dataNodeConfig.DataNodeDirectory, fileBlockMetaData.Guid.ToString());
            string blockFilePath = FSVolume.GetInstance().MoveTempFileToFreeDirectory(tmpFile, fileBlockMetaData.Guid);
                //FsVolume.MoveTempFileToFreeDirectory(tmpFile, fileBlockMetaData.Guid);
            //System.IO.File.Move(tmpFile,blockFilePath);
            //处理fileBlockMetaData ，即保存下来
            string metaDataPath = blockFilePath + ".meta";
            var stream = System.IO.File.Create(metaDataPath);
            fileBlockMetaData.WriteToBinary(new BinaryWriter(stream));
            stream.Close();
        }

        private void ReadBlockMetaData(out FileBlockMetaData fileBlockMetaData, Guid guid)
        {
            string blockFilePath = FSVolume.GetInstance().GetFilePath(guid) + ".meta";
                //FsVolume.GetInstance().GetFilePath(guid) + ".meta";
            if (blockFilePath == ".meta")
                LogUtil.LogError(string.Format("Cannot Find Block {0}",guid.ToString()));
            //var blockFilePath = Path.Combine(_dataNodeConfig.DataNodeDirectory, guid.ToString()+".meta");
            fileBlockMetaData=new FileBlockMetaData();
            fileBlockMetaData.ReadFromBinary(new BinaryReader(System.IO.File.OpenRead(blockFilePath)));
        }
        //private void PreReadDataCallBack(IAsyncResult result)
        //{
        //    var state = result.AsyncState as CustomTcpState;
        //    state.AllDone.Set();
        //    state.AllDone.Reset();
        //    var networkStream = state.NetworkStream;
        //    if (networkStream == null || null == state.FileStream) return;
        //    try
        //    {
        //        var recv = state.NetworkStream.EndRead(result);
        //        if (state.Position != 0)
        //        {
        //            state.PreEvent.WaitOne();
        //            state.PreEvent.Reset();
        //            Array.Copy(state.PreBuffer, 0, state.Buffer, 0, recv);
        //        }
        //        state.FileStream.BeginWrite(state.Buffer, 0, recv, WriteCallBack, state);
        //        //state.FileStream.Write(state.Buffer, 0, recv);
        //        //state.FileStream.Flush(true);
        //        state.Position += recv;
        //        if (state.Position == state.FileBlockMetaData.Length)
        //        {
        //            state.PreEvent.WaitOne();   //等待写完成
        //            state.FileStream.Flush(true);
        //            state.FileStream.Close();
        //            LogUtil.LogInfo("FileStream Closed!");
        //            networkStream.Flush();
        //            networkStream.Close();
        //            LogUtil.LogInfo("NetworkStream Closed!");
        //            LogUtil.LogInfo("=======End Read BlockData From NetworkStream！=====");
        //            LogUtil.LogInfo(string.Format("=======Update FileBlockInfo at Path:{0}=====", state.FileBlockMetaData.FullPath));
        //            NameNodeRemoteObject.CheckFIleBlockContainsDataNode(state.FileBlockMetaData.FullPath, state.FileBlockMetaData, DataNodeInfo);
        //        }
        //        else
        //        {
        //            networkStream.BeginRead(state.PreBuffer, 0, state.PreBufferSize, PreReadDataCallBack, state);
        //            state.AllDone.WaitOne();
        //            state.AllDone.Reset();
        //        }
        //    }
        //    catch (Exception ex)
        //    {
        //        LogUtil.LogError(ex);
        //    }

        //}

        //private void WriteCallBack(IAsyncResult result)
        //{
        //    var state = result.AsyncState as CustomTcpState;
        //    Debug.Assert(state != null, "state is null");
        //    state.FileStream.EndWrite(result);
        //    state.PreEvent.Set();
        //}
        /*
        /// <summary>
        /// 读取元数据
        /// </summary>
        /// <param name="state"></param>
        public void ReadMetaData(ref CustomTcpState state)
        {
            LogUtil.LogInfo("=====Begin to read MetaData from another dataNode=====");
            var stream = state.NetworkStream;
            var metaDataLengthBytes = new byte[sizeof(int)];
            int length = stream.Read(metaDataLengthBytes, 0, metaDataLengthBytes.Length);
            int metaDataLength = BitConverter.ToInt32(metaDataLengthBytes, 0);
            LogUtil.LogInfo(String.Format("=====MetaData Length:{0}", length));
            var metaDataStream = new MemoryStream();
            metaDataStream.Seek(0, SeekOrigin.Begin);
            int count = 0;
            var buffer = new byte[Config.DetailConfig.BuffSize];
            try
            {
                while (count < metaDataLength)
                {
                    if (metaDataLength - count < Config.DetailConfig.BuffSize)
                    {
                        length = stream.Read(buffer, 0, metaDataLength - count);
                        metaDataStream.Write(buffer, 0, length);
                        break;
                    }
                    else
                    {
                        length = stream.Read(buffer, 0, buffer.Length);
                        metaDataStream.Write(buffer, 0, length);
                        count += length;
                    }
                }

            }
            catch (Exception ex)
            {
                Console.WriteLine(ex);
                LogUtil.LogError(ex);
            }

            state.FileBlockMetaData = SerializeHelper.GetModel(metaDataStream) as FileBlockMetaData;
            Debug.Assert(state.FileBlockMetaData != null, "state.FileBlockMetaData is null");
            Block.SaveMetaData(_dataNodeConfig, state.FileBlockMetaData, state.FileBlockMetaData.Guid);
            LogUtil.LogInfo(String.Format("FileName:{0} Guid:{1} Offset:{2} Length:{3}", state.FileBlockMetaData.FileName, state.FileBlockMetaData.Guid.ToString(), state.FileBlockMetaData.Offset, state.FileBlockMetaData.Length));
            LogUtil.LogInfo("=====End read MetaData=====");

        }
         */

        public void StopTcpListen()
        {
            while (!_tcpListener.Pending())
            {
                _tcpListener.Stop();
            }
            _backgroundTcpListener.CancelAsync();
        }
    }
}
